package raft

//
// this is an outline of the API that raft must expose to
// the service (or tester). see comments below for
// each of these functions for more details.
//
// rf = Make(...)
//   create a new Raft server.
// rf.Start(command interface{}) (index, term, isleader)
//   start agreement on a new log entry
// rf.GetState() (term, isLeader)
//   ask a Raft for its current term, and whether it thinks it is leader
// ApplyMsg
//   each time a new entry is committed to the log, each Raft peer
//   should send an ApplyMsg to the service (or tester)
//   in the same server.
//

import (
	"context"
	"fmt"
	"math/rand"
	"sync"
	"sync/atomic"
	"time"

	"../labrpc"
)

// import "bytes"
// import "../labgob"

//
// as each Raft peer becomes aware that successive log entries are
// committed, the peer should send an ApplyMsg to the service (or
// tester) on the same server, via the applyCh passed to Make(). set
// CommandValid to true to indicate that the ApplyMsg contains a newly
// committed log entry.
//
// in Lab 3 you'll want to send other kinds of messages (e.g.,
// snapshots) on the applyCh; at that point you can add fields to
// ApplyMsg, but set CommandValid to false for these other uses.
//
type ApplyMsg struct {
	CommandValid bool
	Command      interface{}
	CommandIndex int
}

const (
	Follow = iota
	Candidate
	Leader
)

const (
	// 超时基数 300，超时延迟为 300~400
	ElecTimeout = 300

	// Leader 发送心跳时间间隔
	RequestSendGrap = 150 * time.Millisecond

	// RPC超时时间
	RPCTimeout = 30 * time.Millisecond
)

//
// A Go object implementing a single Raft peer.
//
type Raft struct {
	mu        sync.Mutex          // Lock to protect shared access to this peer's state
	peers     []*labrpc.ClientEnd // RPC end points of all peers
	persister *Persister          // Object to hold this peer's persisted state
	me        int                 // this peer's index into peers[]
	dead      int32               // set by Kill()

	// Your data here (2A, 2B, 2C).
	// Look at the paper's Figure 2 for a description of what
	// state a Raft server must maintain.

	// 当前服务器所处于的选举阶段（Follow, Candidate, Leader)
	status int

	// 当前服务器是否收到其他服务器的请求
	reqCh chan int
	// 超时重置标志
	timeReset bool

	// 服务器当前任期
	currentTerm int
	// 在当前任期中收到服务器投票的候选人ID
	votedFor int
	// 服务器日志
	log []Log

	// 将要提交的log的最新索引
	commitIndex int

	// 已经应用到状态机的最新索引
	lastApplied int

	// 集群中所有服务器节点的 commitIndex值的集合
	nextIndex []int

	// 集群中所有节点中已经复制到该服务器的最新日志的索引
	matchIndex []int

	// commit chan
	applyCh chan ApplyMsg
}

type Log struct {
	Term    int
	Command interface{}
}

// return currentTerm and whether this server
// believes it is the leader.
func (rf *Raft) GetState() (int, bool) {

	rf.mu.Lock()
	var term = rf.currentTerm
	var isleader = rf.status == Leader
	// Your code here (2A).
	rf.mu.Unlock()
	return term, isleader
}

//
// save Raft's persistent state to stable storage,
// where it can later be retrieved after a crash and restart.
// see paper's Figure 2 for a description of what should be persistent.
//
func (rf *Raft) persist() {
	// Your code here (2C).
	// Example:
	// w := new(bytes.Buffer)
	// e := labgob.NewEncoder(w)
	// e.Encode(rf.xxx)
	// e.Encode(rf.yyy)
	// data := w.Bytes()
	// rf.persister.SaveRaftState(data)
}

//
// restore previously persisted state.
//
func (rf *Raft) readPersist(data []byte) {
	if data == nil || len(data) < 1 { // bootstrap without any state?
		return
	}
	// Your code here (2C).
	// Example:
	// r := bytes.NewBuffer(data)
	// d := labgob.NewDecoder(r)
	// var xxx
	// var yyy
	// if d.Decode(&xxx) != nil ||
	//    d.Decode(&yyy) != nil {
	//   error...
	// } else {
	//   rf.xxx = xxx
	//   rf.yyy = yyy
	// }
}

//
// the service using Raft (e.g. a k/v server) wants to start
// agreement on the next command to be appended to Raft's log. if this
// server isn't the leader, returns false. otherwise start the
// agreement and return immediately. there is no guarantee that this
// command will ever be committed to the Raft log, since the leader
// may fail or lose an election. even if the Raft instance has been killed,
// this function should return gracefully.
//
// the first return value is the index that the command will appear at
// if it's ever committed. the second return value is the current
// term. the third return value is true if this server believes it is
// the leader.
//
func (rf *Raft) Start(command interface{}) (int, int, bool) {

	rf.mu.Lock()
	defer rf.mu.Unlock()
	if rf.status != Leader {
		return -1, -1, false
	}
	term := rf.currentTerm
	rf.log = append(rf.log, Log{
		Term:    term,
		Command: command,
	})
	index := len(rf.log) - 1
	rf.matchIndex[rf.me] = index
	rf.nextIndex[rf.me] = index + 1
	rf.Fprint(fmt.Sprintf("----------start log----------\n%v\n", rf.log))
	//fmt.Printf("%v\n", Log{
	//	Term: term,
	//	Command: command,
	//})
	//index =

	// Your code here (2B).

	return index, term, true
}

//
// the tester doesn't halt goroutines created by Raft after each test,
// but it does call the Kill() method. your code can use killed() to
// check whether Kill() has been called. the use of atomic avoids the
// need for a lock.
//
// the issue is that long-running goroutines use memory and may chew
// up CPU time, perhaps causing later tests to fail and generating
// confusing debug output. any goroutine with a long-running loop
// should call killed() to check whether it should stop.
//
func (rf *Raft) Kill() {
	atomic.StoreInt32(&rf.dead, 1)
	// Your code here, if desired.
}

func (rf *Raft) killed() bool {
	z := atomic.LoadInt32(&rf.dead)
	return z == 1
}

func (rf *Raft) Server() {
	for {
		timeout := time.Duration(ElecTimeout+rand.Int63n(100)) * time.Millisecond
		ctx, cancel := context.WithTimeout(context.Background(), timeout)
		select {
		case <-ctx.Done():
			rf.startVote()
		case <-rf.reqCh:
			rf.mu.Lock()
			cancel()
			rf.timeReset = true
			rf.mu.Unlock()
		}
	}
}

func (rf *Raft) updateCommitIndex(commitIndex int) {
	rf.commitIndex = commitIndex

	if rf.commitIndex > rf.lastApplied {
		entries := make([]Log, rf.commitIndex-rf.lastApplied)
		copy(entries, rf.log[rf.lastApplied+1:rf.commitIndex+1])

		go func(startIndex int, entries []Log) {
			for i := 0; i < len(entries); i++ {
				newAM := ApplyMsg{
					CommandValid: true,
					Command:      entries[i].Command,
					CommandIndex: startIndex + i,
				}
				rf.applyCh <- newAM
				rf.mu.Lock()
				if rf.lastApplied < startIndex+i {
					rf.lastApplied = startIndex + i
				}
				rf.mu.Unlock()
			}
		}(rf.lastApplied+1, entries)
	}
}

func (rf *Raft) startVote() {
	if rf.killed() {
		return
	}
	time.Sleep(10 * time.Millisecond)
	rf.mu.Lock()
	if rf.status == Follow {
		rf.status = Candidate
	}
	rf.currentTerm++
	peersCount := len(rf.peers)
	agreeCount := 1
	voteOver := false
	rf.votedFor = rf.me
	voteChan := make(chan *RequestVoteReply, 3)
	reqArgs := &RequestVoteArgs{
		Term:        rf.currentTerm,
		CandidateId: rf.me,
	}
	reqArgs.LastLogIndex = len(rf.log) - 1
	reqArgs.LastLogTerm = rf.log[len(rf.log)-1].Term
	var mu sync.Mutex
	for i, _ := range rf.peers {
		if i == rf.me {
			continue
		}
		go func(server int) {
			reqReply := &RequestVoteReply{}
			if rf.sendRequestVote(server, reqArgs, reqReply) {
				mu.Lock()
				if !voteOver {
					voteChan <- reqReply
				}
				mu.Unlock()
			}
		}(i)
	}
	timeout := make(chan int, 1)
	go func() {
		time.Sleep(RPCTimeout)
		timeout <- 1
	}()
	rf.mu.Unlock()
	for {
		select {
		case reqReply := <-voteChan:
			rf.mu.Lock()
			if reqReply.Term > rf.currentTerm {
				rf.status = Follow
				rf.votedFor = -1
				rf.mu.Unlock()
				goto TIMEOUT
			}
			if reqReply.VoteGranted {
				agreeCount++
			}
			rf.mu.Unlock()
		case <-timeout:
			mu.Lock()
			close(voteChan)
			mu.Unlock()
			goto TIMEOUT
		}
	}
TIMEOUT:
	mu.Lock()
	voteOver = true
	mu.Unlock()
	rf.mu.Lock()
	if rf.status == Follow {
		rf.mu.Unlock()
		return
	}
	if agreeCount*2 > peersCount {
		rf.status = Leader
		rf.Fprint("-------i am leader------\n")
		rf.votedFor = -1
		rf.mu.Unlock()
		rf.heartBeat()
	} else {
		rf.mu.Unlock()
	}
}

func (rf *Raft) heartBeat() {
	time.Sleep(10 * time.Millisecond)
	for {
		if rf.killed() {
			return
		}

		rf.mu.Lock()
		if rf.status != Leader {
			rf.mu.Unlock()
			return
		}
		appendChan := make(chan *AppendEntriesReply, 3)
		//peersCount := len(rf.peers)
		//agreeCount := 1
		var mu sync.Mutex
		var beatOver = false
		for i, _ := range rf.peers {
			if i == rf.me {
				continue
			}
			reqArgs := &AppendEntriesArgs{
				Term:         rf.currentTerm,
				LeaderId:     rf.me,
				LeaderCommit: rf.commitIndex,
			}
			reqArgs.PrevLogIndex = rf.nextIndex[i] - 1
			reqArgs.PrevLogTerm = rf.log[reqArgs.PrevLogIndex].Term

			logCache := make([]Log, len(rf.log)-reqArgs.PrevLogIndex-1)
			copy(logCache, rf.log[reqArgs.PrevLogIndex+1:])
			reqArgs.Entries = logCache
			rf.Fprint(fmt.Sprintf("----------------send log--------------\n%v\n", logCache))
			go func(server int, PrevLogIndex, EntriesLen int) {
				reqReply := &AppendEntriesReply{}
				if rf.sendAppendEntries(server, reqArgs, reqReply) {
					mu.Lock()
					if !beatOver {
						reqReply.Server = server
						reqReply.PrevLogIndex = PrevLogIndex
						reqReply.EntriesLen = EntriesLen
						appendChan <- reqReply
					}
					mu.Unlock()
				}
			}(i, reqArgs.PrevLogIndex, len(reqArgs.Entries))
		}
		timeout := make(chan int, 1)
		go func() {
			time.Sleep(RPCTimeout)
			timeout <- 1
		}()
		rf.mu.Unlock()
		for {
			select {
			case reqReply := <-appendChan:
				rf.mu.Lock()
				if rf.status != Leader {
					rf.mu.Unlock()
					goto TIMEOUT
				}
				if reqReply.Success {
					// 这里不能直接设为 len(rf.log)是因为，在select等待过程中，可能会有其他的 Start调用
					// 如果直接设置为 len(rf.log)，则等待过程中的 command也会被提交（这些多余的command实际上没有获得大多数服务器的认可）
					rf.matchIndex[reqReply.Server] = reqReply.PrevLogIndex + reqReply.EntriesLen
					rf.nextIndex[reqReply.Server] = rf.matchIndex[reqReply.Server] + 1

					for N := len(rf.log) - 1; N > rf.commitIndex; N-- {
						agreeCount := 0
						for i := 0; i < len(rf.matchIndex); i++ {
							if rf.matchIndex[i] >= N {
								agreeCount++
							}
						}
						if agreeCount*2 > len(rf.peers) {
							rf.updateCommitIndex(N)
							break
						}
					}
				} else {
					if reqReply.Term > rf.currentTerm {
						rf.status = Follow
						rf.votedFor = -1
						rf.mu.Unlock()
						goto TIMEOUT
					} else {
						if rf.nextIndex[reqReply.Server] > 1 {
							rf.nextIndex[reqReply.Server]--
						}
					}
				}
				rf.mu.Unlock()
			case <-timeout:
				mu.Lock()
				beatOver = true
				close(appendChan)
				close(timeout)
				mu.Unlock()
				goto TIMEOUT
			}
		}
	TIMEOUT:
		rf.mu.Lock()
		if rf.status != Leader {
			rf.mu.Unlock()
			return
		}

		//if agreeCount*2 > peersCount {
		//	//rf.setCommitIndex()
		//	for i := rf.commitIndex + 1; i < len(rf.log); i++ {
		//		newAM := ApplyMsg{
		//			true,
		//			rf.log[i].Command,
		//			i,
		//		}
		//		rf.applyCh <- newAM
		//		rf.Fprint(fmt.Sprintf("commit command %v\n", newAM))
		//		rf.commitIndex = len(rf.log) - 1
		//	}
		//}
		rf.mu.Unlock()
		time.Sleep(RequestSendGrap)
	}
}

//
// the service or tester wants to create a Raft server. the ports
// of all the Raft servers (including this one) are in peers[]. this
// server's port is peers[me]. all the servers' peers[] arrays
// have the same order. persister is a place for this server to
// save its persistent state, and also initially holds the most
// recent saved state, if any. applyCh is a channel on which the
// tester or service expects Raft to send ApplyMsg messages.
// Make() must return quickly, so it should start goroutines
// for any long-running work.
//
func Make(peers []*labrpc.ClientEnd, me int,
	persister *Persister, applyCh chan ApplyMsg) *Raft {
	rf := &Raft{}
	rf.mu.Lock()
	rf.peers = peers
	rf.persister = persister
	rf.me = me
	rf.votedFor = -1
	rf.reqCh = make(chan int, 1)
	rf.status = Follow
	rf.applyCh = applyCh
	rf.commitIndex = 0
	rf.lastApplied = 0

	rf.nextIndex = make([]int, len(rf.peers))
	rf.matchIndex = make([]int, len(rf.peers))
	rf.log = make([]Log, 1)
	rf.log[0] = Log{
		Term:    -1,
		Command: -1,
	}

	for i := 0; i < len(rf.peers); i++ {
		rf.nextIndex[i] = 1
	}

	//_ = os.Remove(fmt.Sprintf("debugInfo_%d", rf.me))
	rf.mu.Unlock()
	// Your initialization code here (2A, 2B, 2C).

	go rf.Server()

	// initialize from state persisted before a crash
	rf.readPersist(persister.ReadRaftState())

	return rf
}
